package simpledb;
import java.util.*;

/**
 * The Join operator implements the relational join operation.
 */
public class Join extends AbstractDbIterator {


    JoinPredicate predicate ;
    DbIterator dbIterator1 ;
    DbIterator dbIterator2 ;
    Tuple tuple1 = null ;


    /**
     * Constructor.  Accepts to children to join and the predicate
     * to join them on
     *
     * @param p The predicate to use to join the children
     * @param child1 Iterator for the left(outer) relation to join
     * @param child2 Iterator for the right(inner) relation to join
     */
    public Join( JoinPredicate p, DbIterator child1, DbIterator child2) {
        // some code goes here

        predicate = p ;
        dbIterator1 = child1 ;
        dbIterator2 = child2 ;
     }

    /**
     * @see TupleDesc#combine(TupleDesc, TupleDesc) for possible implementation logic.
     */
    public TupleDesc getTupleDesc() {

        // some code goes here
        return TupleDesc.combine( dbIterator1.getTupleDesc() , dbIterator2.getTupleDesc() ) ;
    }

    public void open()
        throws DbException, NoSuchElementException, TransactionAbortedException {
        // some code goes here

        dbIterator1.open();
        dbIterator2.open();
    }

    public void close() {
        // some code goes here
    }

    public void rewind() throws DbException, TransactionAbortedException {
        // some code goes here

        tuple1 = null ;
        dbIterator1.rewind();
        dbIterator2.rewind();
    }

    /**
     * Returns the next tuple generated by the join, or null if there are no more tuples.
     * Logically, this is the next tuple in r1 cross r2 that satisfies the join
     * predicate.  There are many possible implementations; the simplest is a
     * nested loops join.
     * <p>
     * Note that the tuples returned from this particular implementation of
     * Join are simply the concatenation of joining tuples from the left and
     * right relation. Therefore, if an equality predicate is used 
     * there will be two copies of the join attribute
     * in the results.  (Removing such duplicate columns can be done with an
     * additional projection operator if needed.)
     * <p>
     * For example, if one tuple is {1,2,3} and the other tuple is {1,5,6},
     * joined on equality of the first column, then this returns {1,2,3,1,5,6}.
     *
     * @return The next matching tuple.
     * @see JoinPredicate#filter
     */
    protected Tuple readNext() throws TransactionAbortedException, DbException {
        // some code goes here

        if(  dbIterator1.hasNext() && tuple1 == null ){
           tuple1 =  dbIterator1.next();
        }

        while( dbIterator1.hasNext() || tuple1 != null  ) {

            while (dbIterator2.hasNext()) {
                Tuple tuple2 = dbIterator2.next();
                System.out.println(tuple1);
                System.out.println(tuple2);
                if (predicate.filter(tuple1, tuple2)) {

                    Tuple tuple = new Tuple(getTupleDesc());
                    for (int i = 0; i < tuple.getTupleDesc().numFields(); i++) {
                        if (i < tuple1.getTupleDesc().numFields()) {
                            tuple.setField(i, tuple1.getField(i));
                        } else {
                            tuple.setField(i, tuple2.getField(i - tuple1.getTupleDesc().numFields()));
                        }
                    }
                    System.out.println("combine  :::" + tuple);
                    return tuple;
                }
                System.out.println("--------------------------");
            }

            if( dbIterator1.hasNext() ) {
                tuple1 =  dbIterator1.next();
            }else{
                tuple1 = null ;
            }

            dbIterator2.rewind();
        }

        return null;
    }


}
